package test.mllib.correlation;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.mllib.linalg.Matrix;
import org.apache.spark.mllib.linalg.Vector;
import org.apache.spark.mllib.linalg.Vectors;
import org.apache.spark.mllib.stat.Statistics;

public class PearsonTest2 {

	public static void main(String[] args) {
		SparkConf sparkConf = new SparkConf().setAppName("LinearRegressionTest").setMaster("local[1]");
		JavaSparkContext sc = new JavaSparkContext(sparkConf);
		JavaRDD<String> data = sc.textFile("..\\..\\data\\spark\\mllib\\LinearRegressionTest.txt");

		JavaRDD<Vector> parsedData = data.map(new Function<String, Vector>() {
	        private static final long serialVersionUID = 1L;
			@Override
			public Vector call(String line) throws Exception {
			    String[] xnStr = line.split(",")[1].split(" ");
				double[] xn = new double[xnStr.length];
				for (int i = 0; i < xn.length; i++) {
					xn[i] = Double.parseDouble(xnStr[i]) / 10000;
				}
				return Vectors.dense(xn);
			}
		}).cache();
		// spearman | pearson
		Matrix correlMatrix = Statistics.corr(parsedData.rdd(), "pearson");
		System.out.println(correlMatrix.toString());
		System.out.println(correlMatrix.apply(0, 1));
		
	}
}
